deps: bump DataFusion 55.0 and Arrow/Parqet 59.2 - #5262
Conversation
|
I had a similar effort here - #4888 Happy to close mine though |
|
Hi @mbutrovich @andygrove is this work still active? It looks like the latest DF comes with several improvements related to nested Parquet projection and selective decoding (some are in 55 and some are in main), which will be quite useful for Comet when processing Parquet nested data. |
|
I'll resolve conflicts |
|
2 metrics tests failed. looks like they come from #5423 @sunchao are you ok to ignore those tests for now and adapt to DF 55 in follow up PR. the |
Yes it's OK for me. I can open a follow-up PR to fix this once the DF 55 PR is merged. |
Squashed rebase of the df55 branch onto upstream/main (525c05b): DataFusion 54.1.0 -> 55.0.0, arrow 58.4.0 -> 59.2.0, iceberg-rust rev 3d84c81 -> 16badac3. Conflicts resolved (union of both sides): - native/spark-expr/src/lib.rs: keep upstream `abs` and df55 `spark_sqrt` math_funcs re-exports. - native/shuffle/src/shuffle_writer.rs: keep upstream `DataFusionError` and df55 `TreeNodeRecursion` / `PhysicalExpr` imports. Ported upstream's newly merged DF54-era code to DF55/arrow59: - rss shuffle writers (writers/rss/*): arrow `CompressionContext` -> `IpcWriteContext`. - PartitionedRankLimitExec (operators/rank_limit.rs): implement the now required `ExecutionPlan::apply_expressions`. Folds in the pre-rebase WIP: iceberg rev bump and two ignored CometAggregateSuite metric tests (df55 peak_mem_used regression follow-up). NOTE: Cargo.lock is taken from df55 as-is. Run `cargo build` / `make core` against DataFusion 55 to regenerate the lock and confirm compilation; this sandbox blocks github.com (iceberg-rust git dep) so it could not be built here.
|
Comet works fine in terms of literals, however for the same input in parquet the Spark has different result @andygrove @sunchao wondering if its expected Spark behavior? |
Blocked until the DataFusion 55.0 crates are published (this branch pins a git commit directly). In the meantime, it doubles as a testbed for DataFusion and upstream-Comet-related fixes that need validation against the new version before it's released, most notably apache/datafusion#24090 below.
Which issue does this PR close?
Closes #4865. Closes #4859.
Rationale for this change
Bumping DataFusion this far ahead of a release requires fixing API breakage and validating that fixes landing upstream in the meantime, like apache/datafusion#24090 (nested Parquet leaf pruning, for #4859), actually work end to end against Comet.
What changes are included in this PR?
Bump DataFusion to a commit past feat: prune unread Parquet leaves when a nested column is cast to a narrower type datafusion#24090, Arrow/Parquet to 59.1.0, and iceberg-rust/opendal to compatible commits (native/Cargo.toml, native/Cargo.lock, native/core/Cargo.toml).
Small API-compat fixes for the new arrow-rs/DataFusion versions:
FixedSizeBinaryArray::fromtotry_from,ShufflePartitioner: Send + SynctoSend, andGroupsAccumulatortrait changes (merge_batchdrops_opt_filter, new requiredconvert_to_state) across the affected agg/array functions, shuffle writers, and execution operators. No behavior change.Add
spark_sqrt, a Comet-only scalar UDF (native/spark-expr/src/math_funcs/sqrt.rs, wired in comet_scalar_funcs.rs and spark/.../serde/math.scala). DataFusion's ownsqrterrors on negative input; Spark'sSqrtreturns NaN. Unrelated to the DataFusion bump, just riding this branch.Comet native Parquet scan reads full nested columns despite pruned Spark ReadSchema #4859: Comet's
SparkPhysicalExprAdapterswaps DataFusion'sCastExprfor Comet's ownCometCastColumnExprwhenever a cast touches a Struct, List, or Map, to apply Spark-specific semantics viaspark_parquet_convert. #24090's leaf-pruning inbuild_projection_read_planonly recognizes the literalCastExprtype, so once Comet swaps it, pruning never fires and the whole column gets read regardless of the DataFusion bump. Same failure shape as perf: unwrap identity casts in schema adapter to enable Parquet stats pruning #4730, which hit an analogous problem with row-group pruning.Fix: add
is_pure_structural_narrowing(native/core/src/parquet/schema_adapter.rs). DataFusion's owndatafusion_common::nested_struct::cast_column, whichColumnarValue::cast_toruns at execution time, already matches struct fields by name, drops extra source fields, and null-fills missing target fields, exactly the shape a pruned nested ReadSchema needs. For a pure structural narrowing (fields dropped only, no leaf-type change, no Map, no Parquet field-id matching, no case-insensitive-only match), that's identical to what Comet's own path computes. When the predicate holds,replace_with_spark_castleaves DataFusion'sCastExprin place instead of swapping inCometCastColumnExpr, so #24090's pruning sees it.Built as an allow list (only Struct/List/LargeList recurse, every leaf must be an exact type match), not a deny list, since a deny list fails open: a future Comet special case not mirrored there would silently produce wrong results instead of just missing an optimization. Map narrowing is still uncovered (
nested_struct::cast_columnhas no Map arm), needs a Comet-side accommodation or an upstream extension.Hoist the two regression tests from fix: [DO NOT MERGE] Validate nested Parquet leaf pruning against datafusion#23391 #4866 (array of struct and array of struct of struct leaf pruning) into CometNativeReaderSuite.
CometTPCHQuerySuite: raise MEMORY_OFFHEAP_SIZE to 4g, unrelated CI tuning found necessary while running this branch's suites.
How are these changes tested?
is_pure_structural_narrowing(allowed: struct and list-of-struct-of-struct narrowing; denied: case-insensitive-only match, missing target field, zero field overlap, Parquet field-id matching at top and nested levels, Map, Dictionary, timestamp NTZ to LTZ relabeling, leaf type change), plus two end-to-end tests through the realPhysicalExprAdapter::rewritepipeline and two tests comparing Comet'sspark_parquet_convertagainst the actual DataFusion functions this relies on for byte-identical output.#4859regression tests in CometNativeReaderSuite pass; full CometNativeReaderSuite passes too, to catch any regression in other array/struct/map shapes from swapping fewer casts.